package qgs.zisu.javaconcurrency.javaconcurrency.chapter13_ThreadPool;

import com.sun.xml.internal.ws.addressing.WsaTubeHelperImpl;
import javafx.concurrent.Task;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.IntStream;

/**
 * @Author: 蔡翔
 * @Date: 2019/10/21 13:28
 * @Version 1.0
 */
public class SimpleThreadPool extends Thread{
    private int size;
    private int queueSize;
    private final static int DEFAULT_SIZE = 10;
    private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();
    private static volatile int seq = 0;
    private final static String THREAD_PREFIX = "SIMPLE_THREAD_POOL-";
    private final static ThreadGroup GROUP = new ThreadGroup("Pool_Group");
    private final static List<WorkerTask> THREAD_QUEUE = new ArrayList<>();
    //默认任务队列
    private final static int DEFAULT_TASK_QUEUE_SIZE = 2000;
    private DiscardPolicy discardPolicy;
    private volatile boolean destory = false;

    public final static DiscardPolicy DEFAULT_DISCARD_POLICY = ()->{
        throw new DiscardException("Discard This Task");
    };
    private int min;
    private int max;
    private int active;

    public SimpleThreadPool(){
        // 这里的this 就是就是下面SimpleThreadPool(int size,int queueSize)
        this(12,4,8,DEFAULT_TASK_QUEUE_SIZE,()->{
            throw new DiscardException("Discard this Task");
        });
    }
    public SimpleThreadPool(int max,int min,int active,int queueSize,DiscardPolicy discardPolicy){
        this.max = max;
        this.min = min;
        this.active = active;
        this.queueSize = queueSize;
        this.discardPolicy = discardPolicy;
        init();
    }
    public interface DiscardPolicy{
        void discard();
    }
    public static class DiscardException extends RuntimeException{
        public DiscardException(String message){
            super(message);
        }
    }

    @Override
    public void run(){
        while (!destory){
            System.out.printf("Pool# min:%d,max:%d,active:%d,current:%d,queneSize:%d\n",
                    this.min,this.max,this.active,this.size,TASK_QUEUE.size());
            try {
                Thread.sleep(5000);
                if(TASK_QUEUE.size()>active && size<active){
                    for (int i=size;i<active;i++){
                        createWorkTask();
                    }
                    System.out.println("The pool increment 4");
                    size = active;
                }else if(TASK_QUEUE.size()>max&& size<max){
                    for(int i = size;i<max;i++){
                        createWorkTask();
                    }
                    System.out.println("The pool increament max");
                    size = max;
                }
                if(TASK_QUEUE.isEmpty() && size>active){
                    System.out.println("========reduce=========");
                    synchronized (TASK_QUEUE){
                        int releaseSize = size -active;
                        //Iterator<> 好处是在遍历的时候，可以remove东西。
                        for(Iterator<WorkerTask> iterator = THREAD_QUEUE.iterator();iterator.hasNext();){
                            if(releaseSize<=0){
                                break;
                            }
                            WorkerTask task = iterator.next();
                            task.close();
                            task.interrupt();
                            iterator.remove();
                            releaseSize--;
                        }
                        size = active;
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void init() {
        for(int i =0;i<this.min;i++){
            createWorkTask();
            // 就是在线程池中 创建size个线程，这些线程是一直开着的，他们会一直成TASK_QUEUE 队列中一个接着一个取任务。
        }
        this.size = min;
        this.start();
    }

    public boolean isDestory(){
        return this.destory;
    }


    public void submit(Runnable runnable){
        if(destory){
            throw new IllegalStateException("The thread pool already destroy and not allow");
        }

        //实际是往TASK_QUEUE 里面插入数据,因为插入的时候有锁，所以写入的时候也应该有锁
        synchronized (TASK_QUEUE){
            TASK_QUEUE.add(runnable);
            TASK_QUEUE.notifyAll();
        }
    }

    private void createWorkTask(){
        WorkerTask task = new WorkerTask(GROUP,THREAD_PREFIX+(seq++));
        task.start();
        THREAD_QUEUE.add(task);
    }

    public void shutdown() throws InterruptedException {
        while (!TASK_QUEUE.isEmpty()){
            Thread.sleep(50);
        }
        int initVal = THREAD_QUEUE.size();
        while (initVal>0){
            for(WorkerTask task:THREAD_QUEUE){
                if(task.getTaskState() == TaskState.BLOCKED){
                    task.interrupt();
                    task.close();
                    initVal--;
                }else {
                    Thread.sleep(10);
                }
            }
        }
        this.destory = true;
        System.out.println("The thread pool disposed.");
    }

    public int getMin() {
        return min;
    }

    public void setMin(int min) {
        this.min = min;
    }

    public int getMax() {
        return max;
    }

    public void setMax(int max) {
        this.max = max;
    }

    public int getActive() {
        return active;
    }

    public void setActive(int active) {
        this.active = active;
    }

    public int getSize() {
        return size;
    }

    public void setSize(int size) {
        this.size = size;
    }

    public int getQueueSize() {
        return queueSize;
    }

    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
    }

    private static class WorkerTask extends Thread{
        //volatile不是一把锁，只不过是内存可见。保证执行顺序正确。
        private volatile TaskState taskState = TaskState.FREE;
        public TaskState getTaskState(){
            return this.taskState;
        }

        public WorkerTask(ThreadGroup threadGroup,String name){
            super(threadGroup,name);
        }

        @Override
        public void run(){
            //注意这个跳转
            OUTER:
            while (this.taskState != TaskState.DEAD){
                synchronized (TASK_QUEUE){
                    while (TASK_QUEUE.isEmpty()){
                        //TASK_QUEUE.isEmpty() 为空就代表当前没有要执行的任务，所以我这个线程池要写wait住，不能死掉，死了后重启会造成性能上损耗。
                        try {
                            // 如果当前没有任务做的时候会阻塞住等待。
                            taskState = TaskState.BLOCKED;
                            TASK_QUEUE.wait();
                        } catch (InterruptedException e) {
                            //这里主要是用来捕捉 外部的 “打断的”
                            // 如果只是单纯的 break，那么只是跳出当前的while循环（一个循环），这是不够的。

                            //如果对线程进行 interrupt() 那么这里会捕捉到异常，并且把这个推到  OUTER: 处
                            break OUTER;
                        }
                    }
                    //移除第一条 来消费。
                    Runnable runnable = TASK_QUEUE.removeFirst();
                    if(runnable != null){
                        taskState = TaskState.RUNNING;
                        //如果不为null。那么来执行第一条任务、
                        runnable.run();
                        taskState = TaskState.FREE;
                    }
                }
            }
        }
        public void close(){
            this.taskState = TaskState.DEAD;
        }
    }

    private enum TaskState{
        FREE,RUNNING,BLOCKED,DEAD
    }

    public static void main(String[] args) throws InterruptedException {
        SimpleThreadPool simpleThreadPool = new SimpleThreadPool(12,4,8,5,SimpleThreadPool.DEFAULT_DISCARD_POLICY);
        IntStream.rangeClosed(0,40)
                .forEach(i->{
                    simpleThreadPool.submit(()->{
                        System.out.println("The runnable :"+i+"be serviced by "+Thread.currentThread().getName());
                        try {
                            Thread.sleep(1_000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("The runnable :"+i+"be serviced by "+Thread.currentThread().getName()+"finished");
                    });
                });
        /*Thread.sleep(10_000);
        simpleThreadPool.shutdown();
        simpleThreadPool.submit(()->{System.out.println("==============");});*/
    }
}
